package com.ljm.boot.kafka.controller;

import com.ljm.boot.kafka.config.KafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @Description
 * @Author Dominick Li
 * @CreateTime 2024/1/16 14:41
 **/
@Slf4j
@RestController
public class KafkaProducerController {

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 正常消息发送
     */
    @GetMapping("/send/{msg}")
    public String sendMessage(@PathVariable String msg) {
       log.info("sendMsg=" + msg);
        kafkaTemplate.send(KafkaConfig.DEFALUT_TOPIC, msg);
        return "success";
    }

    /**
     * 带回调的消息发送
     */
    @GetMapping("/sendCallback/{msg}")
    public String sendCallbackMessage(@PathVariable String msg) {
        kafkaTemplate.send(KafkaConfig.DEFALUT_TOPIC, msg).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.error("send msg to kafka error:{}", throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.info("send msg to kafka success topic={},partition={}，msg={}", result.getRecordMetadata().topic(), result.getRecordMetadata().partition(),result.getProducerRecord().value());
            }
        });
        return "success";
    }

}

